package com.ice.remoting;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

import javax.annotation.Nullable;

import com.alibaba.fastjson.JSON;
import com.ice.framework.common.exception.BusinessException;
import com.ice.remoting.common.RemotingHelper;
import com.ice.remoting.common.ServiceThread;
import com.ice.remoting.data.RemotingCommand;
import com.ice.remoting.exception.*;
import com.ice.remoting.listener.ChannelEventListener;
import com.ice.remoting.netty.*;
import com.ice.remoting.utils.NetUtil;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.lang.Pair;
import cn.hutool.core.util.ObjectUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class NettyRemotingAbstract {

    protected final NettyEventExecutor nettyEventExecutor = new NettyEventExecutor();
    /**
     * 执行处理器Map
     */
    protected final HashMap<Integer/* request code */, Pair<NettyRequestProcessor, ExecutorService>> processorTable =
        new HashMap<>(64);

    /**
     * 默认执行处理器
     */
    protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessorPair;


    /**
     * custom rpc hooks
     */
    protected List<RPCHook> rpcHooks = new ArrayList<>();


    /**
     * 响应请求Map
     */
    protected final ConcurrentMap<Integer /* opaque */, ResponseFuture> responseTable = new ConcurrentHashMap<>(256);

    /**
     * 回调处理线程
     */
    public abstract ExecutorService getCallbackExecutor();

    /**
     * 用户事件监听
     */
    public abstract ChannelEventListener getChannelEventListener();

    public void putNettyEvent(final NettyEvent event) {
        this.nettyEventExecutor.putNettyEvent(event);
    }

    /**
     * 处理请求，根据请求code，查询注册处理器，如果没有，则使用默认处理器，没有则异常
     */
    public void processRequestCommand(ChannelHandlerContext ctx, RemotingCommand miiMessage) {
        Pair<NettyRequestProcessor, ExecutorService> matched =
            this.processorTable.get(NetUtil.byte2int(miiMessage.getCommand()));
        final Pair<NettyRequestProcessor, ExecutorService> pair =
            null == matched ? this.defaultRequestProcessorPair : matched;
        if (pair == null) {
            String error = " request type " + miiMessage.getCommand() + " not supported";
            log.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
            throw new BusinessException(ServerErrorCode.PAIR_NOT_REGISTER);
        }
        Runnable run = buildProcessRequestHandler(ctx, miiMessage, pair);
        try {
            pair.getValue().submit(run);
        } catch (Exception e) {
            log.error("执行任务异常:{}", ExceptionUtil.getMessage(e));
        }
    }

    public Runnable buildProcessRequestHandler(ChannelHandlerContext ctx, RemotingCommand request,
        Pair<NettyRequestProcessor, ExecutorService> pair) {
        return () -> {
            RemotingCommand response = null;
            Exception exception = null;
            try {
                String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                try {
                    doBeforeRpcHooks(ctx.channel(), remoteAddr, request);
                } catch (AbortProcessException e) {
                    throw e;
                } catch (Exception e) {
                    exception = e;
                }
                if (exception == null) {
                    response = pair.getKey().processRequest(ctx, request);
                }

                try {
                    doAfterRpcHooks(ctx.channel(), remoteAddr, request, response);
                } catch (AbortProcessException e) {
                    throw e;
                } catch (Exception e) {
                    exception = e;
                }
                if (exception != null) {
                    throw exception;
                }
                writeResponse(ctx.channel(), request, response);
            } catch (Exception e) {
                log.error("process request exception {}", ExceptionUtil.getRootCauseMessage(e));
                // TODO 异常暂不处理
            }
        };
    }

    public void writeResponse(Channel channel, RemotingCommand request, @Nullable RemotingCommand response) {
        if (request.isOnewayRPC() || ObjectUtil.isEmpty(response)) {
            log.debug("无返回数据");
            return;
        }
        channel.writeAndFlush(response);
    }

    protected void doBeforeRpcHooks(Channel channel, String addr, RemotingCommand request) {
        if (rpcHooks.size() > 0) {
            for (RPCHook rpcHook : rpcHooks) {
                rpcHook.doBeforeRequest(channel, addr, request);
            }
        }
    }

    public void doAfterRpcHooks(Channel channel, String addr, RemotingCommand request, RemotingCommand response) {
        if (rpcHooks.size() > 0) {
            for (RPCHook rpcHook : rpcHooks) {
                rpcHook.doAfterResponse(channel, addr, request, response);
            }
        }
    }

    /**
     * Custom RPC hooks.
     */
    public List<RPCHook> getRPCHook() {
        return rpcHooks;
    }


    public void registerRPCHook(RPCHook rpcHook) {
        if (rpcHook != null && !rpcHooks.contains(rpcHook)) {
            rpcHooks.add(rpcHook);
        }
    }

    public void clearRPCHook() {
        rpcHooks.clear();
    }

    class NettyEventExecutor extends ServiceThread {
        private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<>();

        public void putNettyEvent(final NettyEvent event) {
            int currentSize = this.eventQueue.size();
            int maxSize = 10000;
            if (currentSize <= maxSize) {
                this.eventQueue.add(event);
            } else {
                log.warn("event queue size [{}] over the limit [{}], so drop this event {}", currentSize, maxSize,
                    event.toString());
            }
        }

        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");

            final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();

            while (!this.isStopped()) {
                try {
                    NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
                    if (event != null && listener != null) {
                        switch (event.getType()) {
                            case IDLE:
                                listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CLOSE:
                                listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CONNECT:
                                listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
                                break;
                            case EXCEPTION:
                                listener.onChannelException(event.getRemoteAddr(), event.getChannel());
                                break;
                            case ACTIVE:
                                listener.onChannelActive(event.getRemoteAddr(), event.getChannel());
                                break;
                            default:
                                break;
                        }
                    }
                } catch (Exception e) {
                    log.warn(this.getServiceName() + " service has exception. ", e);
                }
            }

            log.info(this.getServiceName() + " service end");
        }

        @Override
        public String getServiceName() {
            return NettyEventExecutor.class.getSimpleName();
        }
    }

    public boolean processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        final int opaque = cmd.getOpaque();
        final ResponseFuture responseFuture = responseTable.get(opaque);
        if (responseFuture != null) {
            responseFuture.setResponseCommand(cmd);

            responseTable.remove(opaque);

            if (responseFuture.getInvokeCallback() != null) {
                executeInvokeCallback(responseFuture);
            } else {
                responseFuture.putResponse(cmd);
            }
            return true;
        }
        return false;
    }

    public void scanResponseTable() {
        final List<ResponseFuture> rfList = new LinkedList<>();
        Iterator<Map.Entry<Integer, ResponseFuture>> it = this.responseTable.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<Integer, ResponseFuture> next = it.next();
            ResponseFuture rep = next.getValue();
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                it.remove();
                rfList.add(rep);
                log.warn("remove timeout request, {}, {}", rep.getOpaque(), JSON.toJSON(rep.getRequest()));
            }
        }

        for (ResponseFuture rf : rfList) {
            try {
                executeInvokeCallback(rf);
            } catch (Throwable e) {
                log.warn("scanResponseTable, operationComplete Exception", e);
            }
        }
    }

    public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
        final InvokeCallback invokeCallback) {
        invokeImpl(channel, request, timeoutMillis).whenComplete((v, t) -> {
            if (t == null) {
                invokeCallback.operationComplete(v);
            } else {
                ResponseFuture future = new ResponseFuture(channel, request.getOpaque(), request, timeoutMillis, null);
                future.setCause(t);
                invokeCallback.operationComplete(future);
            }
        }).thenAccept(responseFuture -> invokeCallback.operationSucceed(responseFuture.getResponseCommand()))
            .exceptionally(t -> {
                invokeCallback.operationFail(t);
                return null;
            });
    }

    public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis) throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        try {
            return invokeImpl(channel, request, timeoutMillis).thenApply(ResponseFuture::getResponseCommand)
                .get(timeoutMillis, TimeUnit.MILLISECONDS);
        } catch (ExecutionException e) {
            throw new RemotingSendRequestException(channel.remoteAddress().toString(), e.getCause());
        } catch (TimeoutException e) {
            throw new RemotingTimeoutException(channel.remoteAddress().toString(), timeoutMillis, e.getCause());
        }
    }

    protected CompletableFuture<ResponseFuture> invokeImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis) {
        String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
        doBeforeRpcHooks(channel, channelRemoteAddr, request);
        return invoke0(channel, request, timeoutMillis).whenComplete((v, t) -> {
            if (t == null) {
                doAfterRpcHooks(channel, channelRemoteAddr, request, v.getResponseCommand());
            }
        });
    }

    protected CompletableFuture<ResponseFuture> invoke0(final Channel channel, final RemotingCommand request,
        final long timeoutMillis) {
        CompletableFuture<ResponseFuture> future = new CompletableFuture<>();

        int opaque = request.getOpaque();

        AtomicReference<ResponseFuture> responseFutureReference = new AtomicReference<>();
        ResponseFuture responseFuture =
            new ResponseFuture(channel, opaque, request, timeoutMillis, new InvokeCallback() {
                @Override
                public void operationComplete(ResponseFuture responseFuture) {

                }

                @Override
                public void operationSucceed(RemotingCommand response) {
                    future.complete(responseFutureReference.get());
                }
            });
        responseFutureReference.set(responseFuture);
        this.responseTable.put(opaque, responseFuture);
        try {
            channel.writeAndFlush(request).addListener((ChannelFutureListener)f -> {
                if (f.isSuccess()) {
                    responseFuture.setSendRequestOK(true);
                    return;
                }
                // 发送失败处理
                requestFail(opaque);
            });
        } catch (Exception e) {
            responseTable.remove(opaque);
            log.warn("send a request command to channel <{}> Exception,{}",RemotingHelper.parseChannelRemoteAddr(channel), e);
            future.completeExceptionally(
                new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e));
        }
        return future;
    }

    public void invokeOnewayImpl(final Channel channel, final RemotingCommand request)
        throws RemotingSendRequestException {
        request.markOnewayRPC();
        try {
            channel.writeAndFlush(request).addListener((ChannelFutureListener)f -> {
                if (!f.isSuccess()) {
                    log.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
                }
            });
        } catch (Exception e) {
            log.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
            throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
        }
    }


    private void requestFail(final int opaque) {
        ResponseFuture responseFuture = responseTable.remove(opaque);
        if (responseFuture != null) {
            responseFuture.setSendRequestOK(false);
            responseFuture.putResponse(null);
            try {
                executeInvokeCallback(responseFuture);
            } catch (Throwable e) {
                log.warn("execute callback in requestFail, and callback throw", e);
            }
        }
    }

    /**
     * Execute callback in callback executor. If callback executor is null, run directly in current thread
     */
    private void executeInvokeCallback(final ResponseFuture responseFuture) {
        boolean runInThisThread = false;
        ExecutorService executor = this.getCallbackExecutor();
        if (executor != null && !executor.isShutdown()) {
            try {
                executor.submit(() -> {
                    try {
                        responseFuture.executeInvokeCallback();
                    } catch (Throwable e) {
                        log.warn("execute callback in executor exception, and callback throw", e);
                    }
                });
            } catch (Exception e) {
                runInThisThread = true;
                log.warn("execute callback in executor exception, maybe executor busy", e);
            }
        } else {
            runInThisThread = true;
        }

        if (runInThisThread) {
            try {
                responseFuture.executeInvokeCallback();
            } catch (Throwable e) {
                log.warn("executeInvokeCallback Exception", e);
            }
        }
    }
}
